package comm.receivedata;


import com.google.common.collect.Lists;
import com.google.gson.reflect.TypeToken;
import comm.bean.GroupBean;
import comm.config.CommMsgTopic;
import comm.data.KafkaData;
import comm.data.ReadFileClass;
import comm.serverhandle.SocketClass;
import exsun.jt808.data.util.GsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.tomcat.util.buf.HexUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.io.OutputStream;
import java.lang.reflect.Type;
import java.net.Socket;

import java.util.ArrayList;
import java.util.List;

@Component
public class TranspondStorage {
    private static final Logger logger = LoggerFactory.getLogger(TranspondStorage.class);

    @Autowired
    GroupBean groupBean;

    @Autowired
    SocketClass socketClass;

    @Autowired
    ReadFileClass readFileClass;

    //订阅原始数据  CommMsgTopic.rawData
    @KafkaListener(id = "${handlecenter.kafka.consumer.group.id}", topics = "${handlecenter.kafka.consumer.TranspondMessageType.id}", containerFactory = "HandleCenterBatchRecordConsumerFactory")
    private void channelWriteBatchRecordConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        Start(records, ack);
//        try {
//            Socket socket = socketClass.getInstance();
////            Socket socket = getSocket();
//            OutputStream os = socket.getOutputStream();
//            if (!socketClass.isServerClose(socket)) {
//                for (ConsumerRecord<?, ?> record : records) {
//                    String msg = record.value().toString();
//                    Type type = new TypeToken<KafkaData>() {
//                    }.getType();
//
//                    KafkaData dataBodyClass = GsonUtils.getObjectFromJson(msg, type);
//                    //由于接收的消息体类型和发送的不一致，这里是强制把类中Object的类型转成byte[]
//                    ArrayList<String> list = Lists.newArrayList(readFileClass.getReadFile());
//                    if (list.contains(dataBodyClass.getDevNo())) {
//                        byte[] data = dataBodyClass.getObject();
//                        logger.info("设备号：" + dataBodyClass.getDevNo() + ",转发的原始数据，byte[]转Hex后的数据：" + HexUtils.toHexString(data));
//                        os.write(data, 0, data.length);
//                        os.flush();
//                    }
//                }
//            } else {
////                logger.info("转发数据端口未开启！！！，请检查！！！");
//
//            }
//            ack.acknowledge();
//        } catch (Exception ex) {
//            logger.error("异常："+ex.toString());
//        }
    }

    //转发所有设备版本
    public void Start(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        Socket socket = socketClass.getInstance();
        try {
//            Socket socket = getSocket();

            if (socketClass.isServerClose(socket)) {
                for (ConsumerRecord<?, ?> record : records) {
                    String msg = record.value().toString();
                    Type type = new TypeToken<KafkaData>() {
                    }.getType();

                    KafkaData dataBodyClass = GsonUtils.getObjectFromJson(msg, type);
                    //由于接收的消息体类型和发送的不一致，这里是强制把类中Object的类型转成byte[]
                    ArrayList<String> list = Lists.newArrayList(readFileClass.getReadFile());
                    if (list.contains(dataBodyClass.getDevNo())) {
                        byte[] data = dataBodyClass.getObject();
                        logger.info("设备号：" + dataBodyClass.getDevNo() + ",转发的原始数据，byte[]转Hex后的数据：" + HexUtils.toHexString(data));
                        OutputStream os = socket.getOutputStream();
                        os.write(data, 0, data.length);
                        os.flush();
                    }
                }
            } else {
//                logger.info("转发数据端口未开启！！！，请检查！！！");
                logger.info("正在尝试重连！！！，当前状态：" + socketClass.isServerClose(socket) + "，true：连接正常，false：连接出现异常");
                Start(records, ack);
            }
            ack.acknowledge();
        } catch (Exception ex) {
            //如有此出打印，可能由于网络传输不稳定导致
            logger.error("订阅原始数据处理时出现异常，信息如下：" + ex.toString()+"，通讯状态"+socketClass.isServerClose(socket));
        }
    }
}
